package com.bw.gmall.realtime.common.util;

import com.bw.gmall.realtime.common.constant.Constant;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Properties;

/**
 和Kafka交互要用到Flink提供的KafkaSource及KafkaSink，
 Flink-1.16开始，原先用于交互的FlinkKafkaConsumer、
 FlinkKafkaProducer被标记为过时，不推荐使用。为了提高模板代码的复用性，
 将KafkaSource的构建封装到FlinkSourceUtil工具类中。
 */
public class FlinkSourceUtil {
    /**
     * kafka消费者
     * @param groupId
     * @param topic
     * @return
     */
    public static KafkaSource<String> getKafkaSource(String groupId,
                                                     String topic) {
        return KafkaSource.<String>builder()
                .setBootstrapServers(Constant.KAFKA_BROKERS)
                .setGroupId(groupId)
                .setTopics(topic)
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new DeserializationSchema<String>() {
                    @Override
                    public String deserialize(byte[] message) throws IOException {
                        if (message != null) {
                            return new String(message, StandardCharsets.UTF_8);
                        }
                        return null;
                    }

                    @Override
                    public boolean isEndOfStream(String nextElement) {
                        return false;
                    }

                    @Override
                    public TypeInformation<String> getProducedType() {
                        return Types.STRING;
                    }
                })
                .build();
    }

    /**
     * cdc
     * @param processDataBase
     * @param processDimTableName
     * @return
     */
        public static MySqlSource<String> getMysqlSource(String processDataBase, String processDimTableName) {
            Properties props =new Properties();
            props.setProperty("useSSL", "false");
            props.setProperty("allowPublicKeyRetrieval", "true");

            return MySqlSource.<String>builder()
                    .jdbcProperties(props)
                    .hostname(Constant.MYSQL_HOST)
                    .port(Constant.MYSQL_PORT)
                    .databaseList(Constant.PROCESS_DATABASE) // monitor all tables under inventory database
                    .username(Constant.MYSQL_USER_NAME)
                    .password(Constant.MYSQL_PASSWORD)
                    .tableList(processDataBase+"."+processDimTableName)
                    .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to String
                    .startupOptions(StartupOptions.initial())
                    .build();
        }
    }

